package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TableProcessDwd;
import com.bw.gmall.realtime.common.bean.TrafficHomeDetailPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.function.DorisMapFunction;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsTrafficHomeDetailPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficHomeDetailPageViewWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE,Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW,4,10023);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
       // 1. 过滤, 解析成 pojo 类型

        //elt清洗数据
        SingleOutputStreamOperator<JSONObject> streamOperator = etlStream(dataStreamSource);


        // 根据 mid 分组  看 首页 和 商品详细页面的 独立访客 所以需要两个 状态 计算 相应的 uv
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> streamOperator1 = CountPVStream(streamOperator);

        // 2. 开窗聚合

        //水位线
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> waterStream = waterStream(streamOperator1);

        //进行 开窗聚合
        //首先根据 维度 进行 分组

        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> streamOperator2 = WindowStream(waterStream);


        streamOperator2.print();
        // 3. 写出到 doris
        streamOperator2.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE+".dws_traffic_home_detail_page_view_window",Constant.DWS_TRAFFIC_HOME_DETAIL_PAGE_VIEW_WINDOW));



    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> WindowStream(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> waterStream) {
         return waterStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean t1, TrafficHomeDetailPageViewBean t2) throws Exception {
                        //滚动 窗口  没有重叠数据，如果是 滑动窗口就要考虑 数据重合问题，要进行 new

                        t1.setHomeUvCt(t1.getHomeUvCt() + t2.getHomeUvCt());
                        t1.setGoodDetailUvCt(t1.getGoodDetailUvCt() + t2.getGoodDetailUvCt());

                        return t1;
                    }
                }, new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow timeWindow, Iterable<TrafficHomeDetailPageViewBean> iterable, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {

                        long start = timeWindow.getStart();

                        long end = timeWindow.getEnd();
                        String tsToDate = DateFormatUtil.tsToDate(System.currentTimeMillis());
                        //获取 iterable 数据

                        Iterator<TrafficHomeDetailPageViewBean> iterator = iterable.iterator();

                        TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean = iterator.next();

                        trafficHomeDetailPageViewBean.setStt(DateFormatUtil.tsToDateTime(start));

                        trafficHomeDetailPageViewBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        trafficHomeDetailPageViewBean.setCurDate(tsToDate);

                        collector.collect(trafficHomeDetailPageViewBean);
                    }
                });
    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> waterStream(SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> streamOperator1) {
         return streamOperator1.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficHomeDetailPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner((event, timestamp) -> event.getTs()));
    }

    private static SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> CountPVStream(SingleOutputStreamOperator<JSONObject> streamOperator) {
        return streamOperator.keyBy(x -> x.getJSONObject("common").getString("mid")).process(new ProcessFunction<JSONObject, TrafficHomeDetailPageViewBean>() {
            //看 首页 和 商品详细页面的 独立访客 所以需要两个 状态
            ValueState<String> state1;

            ValueState<String> state2;

            @Override
            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("state", String.class);
                state1 = getRuntimeContext().getState(stateDescriptor);


                ValueStateDescriptor<String> stateDescriptor1 = new ValueStateDescriptor<>("state", String.class);
                state2 = getRuntimeContext().getState(stateDescriptor1);

            }


            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TrafficHomeDetailPageViewBean>.Context context, Collector<TrafficHomeDetailPageViewBean> collector) throws Exception {

                JSONObject common = jsonObject.getJSONObject("common");

                JSONObject page = jsonObject.getJSONObject("page");

                String pageId = page.getString("page_id");


                Long ts = jsonObject.getLong("ts");

                String tsToDate = DateFormatUtil.tsToDate(ts);


                String value1 = state1.value();

                String value2 = state2.value();

                //首页 独立访客数
                long honect = 0L;

                //商品详细 独立访客数
                long goodct = 0L;

                //看到 首页 独立访客 和 商品详细页面的独立访客

                if ("home".equals(pageId) && !tsToDate.equals(value1)) {

                    honect = 1L;

                    state1.update(tsToDate);

                } else if ("good_detail".equals(pageId) && !tsToDate.equals(value2)) {
                    goodct = 1L;

                    state2.update(tsToDate);

                }

                if (honect + goodct == 1) {
                    collector.collect(TrafficHomeDetailPageViewBean.builder()
                            .homeUvCt(honect)
                            .ts(ts)
                            .goodDetailUvCt(goodct)
                            .curDate(tsToDate)
                            .build());
                }
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> etlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                if (s != null) {

                    try {
                        JSONObject jsonObject = JSON.parseObject(s);

                        if (jsonObject != null) {

                            JSONObject common = jsonObject.getJSONObject("common");
                            JSONObject page = jsonObject.getJSONObject("page");
                            if (common != null && page != null) {

                                String mid = common.getString("mid");

                                if (mid != null) {

                                    collector.collect(jsonObject);
                                }
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }
}
